package com.zp.self.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

@Component
public class KafkaListenerMy {
//    @EventListener(condition = "#event.msgs.topic()=='test'")
//    public void rocketmqMsgListener3(MessageEvent event) {
//        ConsumerRecord msgs = event.getMsgs();
//        log.info("监听到了Canal发送到Test 的数据修改............." + msgs);
//    }

    //------------------- 单个消费 -------------------
//    @KafkaListener(topics = {"log"},groupId = "system_log", id="order001",topicPartitions = {@TopicPartition(topic = "T6",partitions = {"0","1"},partitionOffsets = @PartitionOffset(partition = "2",initialOffset = "100"))})
//    public void listen(ConsumerRecord<?, ?> record) {
//        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
//        if (kafkaMessage.isPresent()) {
//            Object message = kafkaMessage.get();
//            log.info("监听到了Canal发送到Test 的数据修改............." + message);
//        }
//    }
    @KafkaListener(topics = {"T6"},groupId = "system_log"/*,id="order002",topicPartitions = {@TopicPartition(topic = "T6",partitions = {"2"})}*/)
    public void listen2(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            System.out.println(message);
        }
    }
}